# Чтение сообщений

Отличительной особенностью релиза >= 1.1.0 является возможность чтения сообщений из нескольких топиков в рамках одного консьюмера. Такой подход позволяет сэкономить память, но немного снижает параллельность, если после чтения сообщения выполняется бизнес логика обработки сообщения. Рекомендуется использовать именно такой подход.

Возможность создавать консьюмера на каждый топик также остается. В этом случае повышается параллельность работы, но есть нюансы с логированием информации от консьюмеров в файлы (пишется информация от последнего подключенного консьюмера).

## Последовательность использования методов компоненты и их описание

### Установка параметра group.id

Минимальный параметр, который требуется установить, для того чтобы начать получать сообщения - это **group.id**

```1c
УстановитьПараметр("group.id", ИмяВашейГруппы);
```

**ИмяВашейГруппы** - Строка.

### Установка позиции чтения

В большинстве случаев отправной точкой - откуда начинать считывать сообщения консьюмеру - управлять не нужно.
По умолчанию консьюмер начнет потреблять сообщения с последнего зафиксированного смещения. Если для темы + раздела и группы нет ранее зафиксированного смещения, он вернется к свойству конфигурации темы **auto.offset.reset**, которое установлено по умолчанию в **latest**.
Таким образом, консьюмер начинает потреблять сообщения с конца раздела (будут потребляться только новые сообщения).

Но бывают ситуации, в которых необходимо заново перечитать сообщения с определенной позиции. Для этого в компоненте реализован следующий метод:

```1c
УстановитьПозициюЧтения(Топик, Смещение, Партиция);
```

**Топик** - Строка. Имя топика.

**Смещение** - Число. Позиция, с которой нужно начать чтение.

**Партиция** - Число. Номер партиции, по умолчанию = 0.

Данный метод **не является обязательным**.

Метод возвращает значение Истина - операция выполнена успешно, Ложь - в противном случае. Текст ошибки можно получить функцией **ПолучитьСообщениеОбОшибке**

### Инициализация консьюмера

```1c
ИнициализироватьКонсьюмера(Брокер, Топики)
```

**Брокер** - Строка. Брокер, например 192.168.0.139:9092 или список брокеров, разделенных запятой. Порт можно не указывать, так же, если используется кластер, то нет необходимости указывать список всех брокеров, достаточно указать основного брокера. Брокер хранит информацию об других, связанных брокерах кластера.

**Топики** - Строка. Список топиков, разделенных запятой

Метод возвращает значение Истина - если инициализация успешно проведена, Ложь - в противном случае. Текст ошибки можно получить функцией **ПолучитьСообщениеОбОшибке**.

Метод является обязательным.


### Установка таймаута ожидания сообщений

Процесс получения сообщений - это по сути бесконечный цикл, внутри которого происходит проверка - а есть ли новые сообщения? Для того чтобы снизить нагрузку на процессор, используется таймаут ожидания, в рамках которого основной поток приложения блокируется на указанное время в ms.

```1c
УстановитьТаймаутОжидания(Таймаут)
```

**Таймаут** - Число. Указывается продолжительность блокировки основного потока приложения в ms. 

Данный метод не является обязательным, в этом случае таймаут будет составлять 500 ms.

Метод возвращает значение Истина - операция выполнена успешно, Ложь - в противном случае. Текст ошибки можно получить функцией **ПолучитьСообщениеОбОшибке**.

Рекомендуется подбирать данный параметр исходя из архитектуры вашего решения и требований к скорости получения сообщений. 

### Получение сообщений

С релиза 1.4.0 появилась возможность использовать новые методы, с помощью которых можно отдельно получать метаданные сообщения, такие как ключ, заголоки и.т.д, а так же тело самого сообщения в виде двоичных данных (как оно существует в источнике) или же строки UTF-8. Смотри [пример](./examples/modern_consumer.md)

```1c
ПрочитатьСообщение()
```

Метод возвращает значение типа булево: 
 + Истина - Было получено очередное сообщение из очереди
 + Ложь - В очереди нет сообщений

Если по результату вернулось значение **Истина**, можно вызывать следующие нижеперечисленные методы:

```1c
ПолучитьДанныеСообщения(ВозвращатьДвоичныеДаннные)
```

**ВозвращатьДвоичныеДаннные** - Булево. Истина - будут возвращены двоичные данные сообщения. Ложь - будет возвращена строка UTF-8. Значение по умолчанию *Ложь*

```1c
ПолучитьКлючСообщения()
```

Метод возвращает строку - ключ сообщения (*key*)

```1c
ПолучитьЗаголовкиСообщения()
```

Метод возвращает JSON строку, которая содержит массив заголовков, каждый заголовок состоит из ключа и значения (key, value)

```1c
ПолучитьТопикСообщения()
```
Метод возвращает топик, из которого было прочитано сообщение (*topic*).

```1c
ПолучитьИдентификаторБрокераСообщения()
```

Метод возвращает идентификатор брокера, из которого было прочитано сообщение (*broker_id*).

```1c
ПолучитьВременнуюМеткуСообщения()
```
Метод возвращает число, unix timestamp в ms. Фактическое время записи сообщения в топик и указанный раздел.

```1c
ПолучитьРазделСообщения()
```
Метод возвращает число. Номер партиции, из которой было прочитано сообщение (*partition*).

```1c
ПолучитьСмещениеСообщения()
```

Метод возвращает число. Текущее смещение сообщения (*offset*).

Данные методы были добавлены с целью повысить отказоустойчивость работы компоненты и дать возможность не получать данные, которые не нужны. 

Кроме использования новых методов - можно использовать старый подход используя метод **Слушать**

```1c
Слушать()
```

Метод возвращает по одному сообщению из буффера прочитанных сообщений. В случае возникновения ошибки возвращает пустую строку. Текст ошибки можно получить функцией **ПолучитьСообщениеОбОшибке**.


Сообщение - строка JSON.
Структура:

+ key - Произвольный ключ сообщения, указанный консьюмером при отправке. Если ключ не был указан - данное свойство не будет добавлено в структуру сообщения
+ headers - Массив заголовков, указанных при отправке. Если заголовки отсутствуют, то данное свойство не будет добавлено в структуру сообщения
+ partition - Номер раздела, из которого произведено чтение
+ offset - Смещение (позиция сообщения в разделе)
+ message - Тело сообщения
+ topic - Топик, из которого было прочитано сообщение
+ timestamp - Unix timestamp в ms. Фактическое время записи сообщения в топик и указанный раздел.
+ broker_id - Идентификатор брокера, из которого было получено сообщение. Данное значение можно использовать для анализа работы кластера.

Пример сообщения:

```json
{
    "key": "a4fc050c-a355-4001-a216-d284aa8f7764",
    "partition": "0",
    "offset": "6576732",
    "message": "{\n  \"id\": \"99999\",\n  \"sum\": \"28827.24\",\n  \"Date\": \"05.10.2023 13:32:39\",\n  \"type\": \"monthly\",\n  \"GUID_Loan\": \"90e2df35-430f-11ed-9ec9-005056130ba6\",\n  \"Method\": \"NewPayment\"\n}\u0006",
    "topic": "testTopic",
    "timestamp": "1696758629706",
    "headers": [
        {
            "key": "level",
            "value": "high"
        },
        {
            "key": "buffer",
            "value": "1024"
        },
        {
            "key": "trusted",
            "value": "true"
        }
    ]
}

```
Метод является обязательным.

### Фиксация смещения после чтения

[Пример кода](./examples/auto_commit_off.md)

В большинстве случаев использовать ручное смещение нет смысла. Информация о прочитанных сообщениях - позиции - фиксируется брокером автоматически. За это отвечают следующие настройки

**enable.auto.commit** = true

**auto.commit.interval.ms** = 5000

В тривиальных сценариях, когда мы прочитали сообщение и сложили его, к примеру, в регистр сведений для дальнейшей обработки - можно оставить автоматическую фиксацию, которая используется по умолчанию. Но в случаях, когда после чтения необходимо выполнить определенную бизнес логику - желательно фиксировать смещения вручную.
При таком подходе, необходимо указать следующую комбинацию параметров

**enable.auto.commit** = false

**enable.auto.offset.store** = false

**enable.partition.eof** = false

```1c
ЗафиксироватьСмещение(Топик, Смещение, Партиция)
```

**Топик** - Строка. 

**Смещение** - Число. Значение смещения прочитанного сообщения + 1

**Партиция** - Число. Номер раздела.

Метод возвращает значение Истина - операция выполнена успешно, Ложь - в противном случае. Текст ошибки можно получить функцией **ПолучитьСообщениеОбОшибке**.

Метод не является обязательным.

### Завершение работы консьюмера 

```1c
ОстановитьКонсьюмера()
```

Метод строго **обязательный**. 

## Пример кода

```1c
	РазрешеноСлушать = ПроверитьЗначениеНастройки();

	Попытка
		Компонента = Новый(СтрШаблон("AddIn.%1.simpleKafka1C", "Integration"));   
	Исключение
		Подключено = ПодключитьВнешнююКомпоненту("ОбщийМакет.КомпонентаSimpleKafka", "Integration", ТипВнешнейКомпоненты.Native, ТипПодключенияВнешнейКомпоненты.Изолированно);
		Если Подключено Тогда
			Компонента = Новый(СтрШаблон("AddIn.%1.simpleKafka1C", "Integration"));  	
		КонецЕсли;
	КонецПопытки;
	
	Если Компонента = Неопределено Тогда
		Возврат;	
	КонецЕсли;
	
	Для каждого Параметр_ Из Параметры Цикл
		Компонента.УстановитьПараметр(Строка(Параметр_.Ключ), Строка(Параметр_.Значение));	
	КонецЦикла;   	
	
	Компонента.КаталогЛогов = ПолучитьКаталогЛогов();
	Результат = Компонента.ИнициализироватьКонсьюмера(Брокер, Топики); 

	Если Не Результат Тогда  
		ТекстОшибки = СтрШаблон("Не удалось инициализировать консьюмера для топиков: %1", Топики);  
		ЗаписьЖурналаРегистрации("Интеграция Kafka. Consumer", УровеньЖурналаРегистрации.Ошибка,,, ТекстОшибки);
		Возврат;
	КонецЕсли;  

	// установка таймаута для ожидания сообщений
	Компонента.УстановитьТаймаутОжидания(Таймаут);		               
		
	Пока РазрешеноСлушать Цикл
		
		РазрешеноСлушать = ПроверитьЗначениеНастройки();

		Если Не РазрешеноСлушать Тогда
			Прервать;	
		КонецЕсли;			      
		
		Сообщение = Компонента.Слушать(); 
		Если НЕ ЗначениеЗаполнено(Сообщение) Тогда                 	
			ОписаниеОшибки = Компонента.ПолучитьСообщениеОбОшибке();                                                         
			Если ЗначениеЗаполнено(ОписаниеОшибки) Тогда
				Сообщить(ОписаниеОшибки)
				Компонента = Неопределено;
				Возврат;
			КонецЕсли;
		КонецЕсли;
				
		КлючСообщения = ПолучитьХешКлючСообщения(Сообщение);
		Если Не СообщениеПрочитано(КлючСообщения) Тогда	     
			ЗаписатьСообщениеВОчередьОбработки(); // здесь пишем в РС, к примеру	
		КонецЕсли;

	КонецЦикла;  
	
	Компонента.ОстановитьКонсьюмера();	
	Компонента = Неопределено;
```
